--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 1e5564cfa34adb2dd4fd9e5ec445e6419d239914
Parents : e02e17d
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-01-07T19:51:48-06:00
refactor(tests): formatting
Changes
1 files changed, 32 insertions(+), 19 deletions(-)
Diff
diff --git a/tests/backend/test_telemetry_integration.py b/tests/backend/test_telemetry_integration.py
index 81e7c3fe..0e825225 100644
--- a/tests/backend/test_telemetry_integration.py
+++ b/tests/backend/test_telemetry_integration.py
@@ -10,30 +10,32 @@ def mock_app():
# We create a simple mock object that has the methods/attributes
# needed by process_incoming_telemetry and other telemetry logic.
app = MagicMock(spec=ReticulumMeshChat)
-
+
# Mock database
app.database = MagicMock()
app.database.telemetry = MagicMock()
-
+
# Mock context
app.current_context = MagicMock()
app.current_context.database = app.database
app.current_context.local_lxmf_destination = MagicMock()
app.current_context.local_lxmf_destination.hexhash = "local_hash"
-
+
# Mock reticulum
app.reticulum = MagicMock()
app.reticulum.get_packet_rssi.return_value = -70
app.reticulum.get_packet_snr.return_value = 12.5
app.reticulum.get_packet_q.return_value = 85
-
+
# Mock websocket_broadcast
app.websocket_broadcast = MagicMock()
-
- # Attach the actual method we want to test if possible,
+
+ # Attach the actual method we want to test if possible,
# but since it's an instance method, we might need to bind it.
- app.process_incoming_telemetry = ReticulumMeshChat.process_incoming_telemetry.__get__(app, ReticulumMeshChat)
-
+ app.process_incoming_telemetry = (
+ ReticulumMeshChat.process_incoming_telemetry.__get__(app, ReticulumMeshChat)
+ )
+
return app
@@ -42,11 +44,13 @@ async def test_process_incoming_telemetry_single(mock_app):
source_hash = "source_hash"
location = {"latitude": 50.0, "longitude": 10.0}
packed_telemetry = Telemeter.pack(location=location)
-
+
mock_lxmf_message = MagicMock()
mock_lxmf_message.hash = b"msg_hash"
- mock_app.process_incoming_telemetry(source_hash, packed_telemetry, mock_lxmf_message)
+ mock_app.process_incoming_telemetry(
+ source_hash, packed_telemetry, mock_lxmf_message
+ )
# Verify database call
mock_app.database.telemetry.upsert_telemetry.assert_called()
@@ -78,7 +82,10 @@ async def test_process_incoming_telemetry_stream(mock_app):
# to handle single entries, and on_lxmf_delivery loops over streams.
for entry_source, entry_timestamp, entry_data in entries:
mock_app.process_incoming_telemetry(
- entry_source, entry_data, mock_lxmf_message, timestamp_override=entry_timestamp
+ entry_source,
+ entry_data,
+ mock_lxmf_message,
+ timestamp_override=entry_timestamp,
)
assert mock_app.database.telemetry.upsert_telemetry.call_count == 2
@@ -98,21 +105,27 @@ async def test_telemetry_request_parsing(mock_app):
# We need to mock handle_telemetry_request on the app
mock_app.handle_telemetry_request = MagicMock()
-
+
# Bind on_lxmf_delivery
- mock_app.on_lxmf_delivery = ReticulumMeshChat.on_lxmf_delivery.__get__(mock_app, ReticulumMeshChat)
-
+ mock_app.on_lxmf_delivery = ReticulumMeshChat.on_lxmf_delivery.__get__(
+ mock_app, ReticulumMeshChat
+ )
+
# Mocking dependencies
mock_app.is_destination_blocked.return_value = False
mock_app.current_context.config.telemetry_enabled.get.return_value = True
- mock_app.database.contacts.get_contact_by_identity_hash.return_value = {"is_telemetry_trusted": True}
- mock_app.database.messages.get_lxmf_message_by_hash.return_value = {} # To avoid JSON error
-
+ mock_app.database.contacts.get_contact_by_identity_hash.return_value = {
+ "is_telemetry_trusted": True
+ }
+ mock_app.database.messages.get_lxmf_message_by_hash.return_value = {} # To avoid JSON error
+
# Call it
mock_app.on_lxmf_delivery(mock_lxmf_message)
-
+
# Verify handle_telemetry_request was called
- mock_app.handle_telemetry_request.assert_called_with("736f757263655f686173685f6279746573")
+ mock_app.handle_telemetry_request.assert_called_with(
+ "736f757263655f686173685f6279746573"
+ )
@pytest.mark.asyncio
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────